package streamAPI.operators;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.testng.annotations.Test;
import util.FlinkConstant;


/**
 * @author yue.cao
 * @since 09-28-2020
 */
public class DemoWithMap {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> source = env.fromElements(FlinkConstant._Common.FROM_0_TO_10);
		source.map(new MapFunction<Integer, Integer>() {
			@Override
			public Integer map(Integer integer) throws Exception {
				return integer * 2;
			}
		}).print();
		env.execute(DemoWithMap.class.getName());
	}

	@Test
	private void t1() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> source = env.fromElements(FlinkConstant._Common.FROM_0_TO_10);
		source.map(i -> i * 2).returns(Integer.class).print();
		env.execute();
	}

	@Test
	private void t2() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> source = env.fromElements(FlinkConstant._Common.FROM_0_TO_10);
		source.map(new RichMapFunction<Integer, Integer>() {

			// 在构造方法之后  map方法之前 执行一次 configuration可以拿到全局配置
			@Override
			public void open(Configuration configuration) throws Exception {
				super.open(configuration);
			}

			// 销毁之前执行一次
			@Override
			public void close() throws Exception {
				super.close();
			}

			@Override
			public Integer map(Integer in) throws Exception {
				return in * 2;
			}
		}).print();
		env.execute();
	}
}
